package realtime.app.func;

import com.alibaba.fastjson.JSONObject;
import io.lettuce.core.api.StatefulRedisConnection;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import org.apache.hadoop.hbase.client.AsyncConnection;
import realtime.common.EduConfig;
import realtime.util.HbaseUtil;
import realtime.util.RedisUtil;

import java.util.Collections;
import java.util.concurrent.CompletableFuture;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Supplier;

/**
 * @Create 2023/9/8 22:53
 */
public abstract class DimAsyncFunction<T> extends RichAsyncFunction<T,T> implements DimJoinFunction<T>{
    private String tableName;
    private AsyncConnection hbaseAsyncConn;
    private StatefulRedisConnection<String,String> redisAsyncConn;

    public DimAsyncFunction(String tableName) {
        this.tableName = tableName;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        hbaseAsyncConn=HbaseUtil.getAsyncConnection();
        redisAsyncConn=RedisUtil.getAsyncRedisConnection();
    }

    @Override
    public void close() throws Exception {
        if (hbaseAsyncConn != null) {
            hbaseAsyncConn.close();
        }
        RedisUtil.closeAsyncRedisConn(redisAsyncConn);
    }



    /**
     *
     * @param value 需要异步处理的主流元素
     * @param resultFuture 用来输出最终异步处理后的元素
     * @throws Exception
     */
    @Override
    public void asyncInvoke(T value, ResultFuture<T> resultFuture) throws Exception {
        CompletableFuture
                //supply执行completablefuture任务,支持返回值
                .supplyAsync(new Supplier<JSONObject>() {
                    @Override
                    public JSONObject get() {
                        return RedisUtil.asyncGetDimInfo(redisAsyncConn, tableName + ":" + getKey(value));
                    }
                })
                // thenApply方法表示，第一个任务执行完成后，执行第二个回调方法任务，会将该任务的执行结果，作为入参，传递到回调方法中，并且回调方法是有返回值的
                .thenApplyAsync(new Function<JSONObject, JSONObject>() {
                    @Override
                    public JSONObject apply(JSONObject jsonObject) {
                        if (jsonObject == null) {
                            System.out.println(tableName + "  " + getKey(value) + "  从 hbase 读取");
                            JSONObject dimInfoFromHbaseAsync = HbaseUtil.getDimInfoFromHbaseAsync(hbaseAsyncConn, EduConfig.HBASE_NAMESPACE, tableName, getKey(value));
                            RedisUtil.asyncSetDimInfo(redisAsyncConn, tableName + ":" + getKey(value), dimInfoFromHbaseAsync);
                            return dimInfoFromHbaseAsync;
                        } else {
                            System.out.println(tableName + "  " + getKey(value) + "  从 redis 读取");
                        }
                        return jsonObject;
                    }

                })
                .thenAcceptAsync(new Consumer<JSONObject>() {
                    @Override
                    public void accept(JSONObject jsonObject) {
                        join(value, jsonObject);//在这个抽象方法内补充维度
                        //收集结果
                        resultFuture.complete(Collections.singletonList(value));
                    }
                });
    }

    //如果异步超时,会执行这个方法
    @Override
    public void timeout(T input, ResultFuture<T> resultFuture) throws Exception {
        // 根据自己情况进行处理
        throw new RuntimeException("异步超时:一般是其他原因导致的异步超时, 请检查: \n" +
                "1.检查集群是否都 ok: hdfs redis hbase kafka. \n" +
                "2.检查下 redis 的配置 bind 0.0.0.0 \n" +
                "3.检查用到的6张维度表是否都在,并且每张表都数据. 最好通过 maxwell-bootstrap 同步一下 \n" +
                "4. 找我" +
                "");
    }
}
